package com.hlz.flink.chapter06;

import com.hlz.flink.chapter05.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @author Hongliang Zhu
 * @create 2023-05-08 23:19
 */
public class WaterMarkTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        DataStreamSource<Event> source = env.fromElements(
                new Event("Mary", "./mary", 1000L),
                new Event("Bob", "./hello", 2000L),
                new Event("Alice", "./prod", 3000L),
                new Event("Alice", "./prod2", 4000L),
                new Event("Bob", "./prod", 5000L),
                new Event("Bob", "./prod1", 6000L),
                new Event("Bob", "./prod2", 7000L),
                new Event("Bob", "./prod3", 8000L),
                new Event("Alice", "./prod2", 4000L),
                new Event("Alice", "./prod2", 4000L),
                new Event("Alice", "./prod2", 4000L),
                new Event("Alice", "./prod2", 4000L),
                new Event("Alice", "./prod2", 4000L));

        // 乱序流watermark生成
        SingleOutputStreamOperator<Event> operator = source.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMillis(5)));
        operator.print();
        env.execute();
    }
}
